package com.sunday.canal.adapter.mongo.service;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.sunday.canal.adapter.mongo.constants.LogConstants;
import com.sunday.canal.adapter.mongo.support.DateTimeUtil;
import com.sunday.canal.adapter.mongo.support.EventData;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Slf4j
@Component
public abstract class BaseMongoSyncService {

    /**
     * @Description: 解析
     * @Param: [entries]
     * @Return: java.util.Collection<com.sunday.canal.adapter.mongo.support.EventData>
     * @Author: zsy
     * @Time: 2023/4/12 16:22
     */
    protected Collection<EventData> parseEntry(List<CanalEntry.Entry> entries) {
        List<EventData> result = new ArrayList<>();
        for (CanalEntry.Entry entry : entries) {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = System.currentTimeMillis() - executeTime;

            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN) {
                    CanalEntry.TransactionBegin begin;
                    try {
                        begin = CanalEntry.TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException(STR."parse event has an error , data:\{entry.toString()}", e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    log.info(LogConstants.transaction_format,
                            entry.getHeader().getLogfileName(),
                            entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getExecuteTime(), DateTimeUtil.format(entry.getHeader().getExecuteTime()),
                            entry.getHeader().getGtid(), delayTime);
                    log.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
                    printXAInfo(begin.getPropsList());
                } else if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                    CanalEntry.TransactionEnd end;
                    try {
                        end = CanalEntry.TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e) {
                        throw new RuntimeException(STR."parse event has an error , data:\{entry.toString()}", e);
                    }
                    // 打印事务提交信息，事务id
                    log.info("----------------\n");
                    log.info(" END ----> transaction id: {}", end.getTransactionId());
                    printXAInfo(end.getPropsList());
                    log.info(LogConstants.transaction_format,
                            entry.getHeader().getLogfileName(),
                            entry.getHeader().getLogfileOffset(),
                            entry.getHeader().getExecuteTime(), DateTimeUtil.format(entry.getHeader().getExecuteTime()),
                            entry.getHeader().getGtid(), delayTime);
                }

                continue;
            }

            if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                CanalEntry.RowChange rowChange;
                try {
                    rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e) {
                    throw new RuntimeException(STR."parse event has an error , data:\{entry.toString()}", e);
                }

                CanalEntry.EventType eventType = rowChange.getEventType();

                log.info(LogConstants.row_format,
                        entry.getHeader().getLogfileName(),
                        entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(),
                        entry.getHeader().getTableName(), eventType,
                        entry.getHeader().getExecuteTime(), DateTimeUtil.format(entry.getHeader().getExecuteTime()),
                        entry.getHeader().getGtid(), delayTime);

                if (eventType == CanalEntry.EventType.QUERY || rowChange.getIsDdl()) {
                    log.info(STR."ddl : \{rowChange.getIsDdl()} ,  sql ----> \{rowChange.getSql()}\{LogConstants.SEP}");
                    continue;
                }

                printXAInfo(rowChange.getPropsList());
                String tableName = entry.getHeader().getTableName();
                rowChange.getRowDatasList().stream()
                        .forEach(rowData -> result.add(buildData(tableName, eventType, rowData)));
            }
        }
        return result;
    }

    public void printXAInfo(List<CanalEntry.Pair> pairs) {
        if (pairs == null) {
            return;
        }

        String xaType = null;
        String xaXid = null;
        for (CanalEntry.Pair pair : pairs) {
            String key = pair.getKey();
            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE")) {
                xaType = pair.getValue();
            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID")) {
                xaXid = pair.getValue();
            }
        }

        if (xaType != null && xaXid != null) {
            log.info(STR." ------> \{xaType} \{xaXid}");
        }
    }

    public void printColumn(List<CanalEntry.Column> columns) {
        for (CanalEntry.Column column : columns) {
            StringBuilder builder = new StringBuilder();
            try {
                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
                        || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY")) {
                    // get value bytes
                    builder.append(STR."\{column.getName()} : \{new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8")}");
                } else {
                    builder.append(STR."\{column.getName()} : \{column.getValue()}");
                }
            } catch (UnsupportedEncodingException e) {
            }
            builder.append(STR."    type=\{column.getMysqlType()}");
            if (column.getUpdated()) {
                builder.append(STR."    update=\{column.getUpdated()}");
            }
            builder.append(LogConstants.SEP);
            log.info(builder.toString());
        }
    }

    protected abstract EventData buildData(String tableName, CanalEntry.EventType eventType, CanalEntry.RowData rowData);


}
